package com.love.cloud.common.redis.queue.config;

import com.love.cloud.common.redis.queue.annotation.QueueConsumer;
import com.love.cloud.common.redis.queue.consumer.BaseConsumer;
import com.love.cloud.common.redis.queue.listener.ConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.Map;

/**
 * @author hongdongqu
 */
@Slf4j
@Configuration
public class RedisStreamConfig {
    @Resource
    private RedisTemplate redisTemplate;
    @Resource
    private RedisStreamQueueProperties redisStreamQueueProperties;
    @Resource
    private ApplicationContext applicationContext;


    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> emailListenerContainerOptions(){
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                //block读取超时时间
                .pollTimeout(Duration.ofSeconds(redisStreamQueueProperties.getPollTimeoutSeconds()))
                //count 数量（一次只获取一条消息）
                .batchSize(redisStreamQueueProperties.getBatchSize())
                //序列化规则
                .serializer(stringRedisSerializer)
                .build();
    }

    /**
     * 开启监听器接收消息
     */
    @Bean
    public StreamMessageListenerContainer<String,MapRecord<String,String,String>> emailListenerContainer(RedisConnectionFactory factory,
                                                                                                         StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,String,String>> streamMessageListenerContainerOptions){

        StreamMessageListenerContainer<String,MapRecord<String,String,String>> listenerContainer = StreamMessageListenerContainer.create(factory,
                streamMessageListenerContainerOptions);
        Map<String,ConsumerListener> map = applicationContext.getBeansOfType(ConsumerListener.class);
        for (ConsumerListener item:map.values()) {
            QueueConsumer annotation = item.getClass().getAnnotation(QueueConsumer.class);
            if(null!=annotation){
                //执行消费者注册逻辑
                String groupName= annotation.groupName();
                String topic= annotation.topic();
                String name= annotation.name();
                //消费组名和主题名称不能为空
                if(StringUtils.isNotEmpty(groupName)&&StringUtils.isNotEmpty(topic)&&StringUtils.isNotEmpty(name)){
                    //创建消费者
                    BaseConsumer baseConsumer=new BaseConsumer(topic,groupName,name,annotation.consumerAck(),redisTemplate,item);
                    //如果 流不存在 创建 stream 流
                    if( !redisTemplate.hasKey(topic)){
                        redisTemplate.opsForStream().add(topic, Collections.singletonMap("", ""));
                        log.info("初始化stream {} success",topic);
                    }
                    //创建消费者组
                    try {
                        redisTemplate.opsForStream().createGroup(topic,groupName);
                    } catch (Exception e) {
                        log.info("消费者组 {} 已存在",groupName);
                    }
                    //注册消费者 消费者名称，从哪条消息开始消费，消费者类
                    // > 表示没消费过的消息
                    // $ 表示最新的消息
                    listenerContainer.receive(
                            Consumer.from(groupName,name),
                            StreamOffset.create(topic, ReadOffset.lastConsumed()),
                            baseConsumer
                    );
                }
            }
        }
        listenerContainer.start();
        return listenerContainer;
    }
}
